/*
    Copyright 2010 Barnet Wagman

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 */

package com.norbl.util.s3;

import com.norbl.util.*;
import java.io.*;
import org.jets3t.service.*;
import org.jets3t.service.security.*;
import org.jets3t.service.impl.rest.httpclient.*;
import org.jets3t.service.model.*;
import org.jets3t.service.utils.*;
import java.util.*;

/** Adds objects to S3.  This class tests whether the objects it
 *  adds to S3 actually arrive (i.e. are available) and performs
 *  parallel writes. In principle the latter can speed up uploads.
 *  However, each write thread uses its own connection, so whether there
 *  is a net performance gain is not obvious.
 *
 * @author Barnet Wagman
 */
public class ObjectAdder {

    public static int N_THREADS_DEFAULT = 8;

    private S3Ops sops;

    private AWSCredentials acs;
    private Jets3tProperties jp;

    private long retryIntervalMillis;
    private long maxWaitMillis;
    private int nThreads;  

    private List<S3Bko> soks;
    private List<Throwable> xList;
    
    private ObjectAdder(S3Ops sops,
                       int nThreads) {
        this.sops = sops;
        this.acs = sops.getACS();
        this.jp = sops.getJet3tProperties();
        this.nThreads = nThreads;       
        this.retryIntervalMillis = sops.getRetryIntervalMillis();
        this.maxWaitMillis = sops.getMaxWaitMillis();
        this.xList = new ArrayList<Throwable>();
    }

    private ObjectAdder(S3Ops sops) {
        this(sops,N_THREADS_DEFAULT);
    }

    public static ObjectAdder add(S3Ops sops, List<S3Bko> soks) {

        ObjectAdder oa = new ObjectAdder(sops);
        oa.add(soks);
        return(oa);
    }

        /** This method returns as soon as all adds have been submitted
         *  to add threads. Use {@link @addsInProgress} to determine
         *  when the adds are complete.
         *
         * @param soks
         */
    private void add(List<S3Bko> soks) throws S3AccessException {

        this.soks = soks;
        List<List> groups = NumericUtil.partition(soks, nThreads);

        for ( List g : groups ) {
             Adder a = new Adder(g);
             (new Thread(a)).start();
        }
    }

    public boolean isDone() throws S3AccessException {

        try {
            int nDone = 0;
            for ( S3Bko sok : soks ) {
                if ( sops.objectExists(sok.bucketName,sok.s3Key,true) )
                    ++nDone;
            }

            return(nDone == soks.size());
        }
        catch(S3ServiceException sx) { throw new S3AccessException(sx); }
    }
    
    public void waitUntilDone(long existenceTestInterval) {
        
        while ( !isDone() ) {
            Time.nap(existenceTestInterval);
        }      
    }

    public boolean hasThrowables() {
        synchronized(xList) {
            return( xList.size() > 0 );
        }
    }

    public synchronized List<Throwable> getThrowables() {
        return(xList);
    }

    private class Adder implements Runnable {
               
        List<S3Bko> bkos;
        S3Service s3;
        private Serializer standardSerializer;
        private Serializer gzipSerializer;

        Adder(List<S3Bko> bkos) {
            this.bkos = bkos;
        }

        public void run() {

            try {               
                s3 = new RestS3Service(acs, null,null,jp);
                for ( S3Bko bko : bkos ) {
                    addObject(bko);
                }
            }
            catch(Throwable x) {
                synchronized(xList) { xList.add(x); }
            }
            finally {
                try {
                    if ( (s3 != null) && !s3.isShutdown() ) s3.shutdown();
                }
                catch(Throwable x) {  
                    synchronized(xList) { xList.add(x); }
                }
            }
        }

        private void addObject(S3Bko bko)
            throws InvalidClassException, IOException, ClassNotFoundException,
                   S3ServiceException {

            long t0 = System.currentTimeMillis();
            S3ServiceException s3x;
            boolean replaceS3Service = false;
            do {
                try {
                    addObj(bko,replaceS3Service);
                    return;
                }
                catch(S3ServiceException x) {
                    s3x = x;
                    replaceS3Service = true;
                    Time.nap(retryIntervalMillis);
                }
            }
            while ( !Time.timedOut(t0,maxWaitMillis) );

            throw s3x;
        }

        private void addObj(S3Bko bko,
                            boolean replaceS3Service)
            throws IOException, S3ServiceException {

            if ( replaceS3Service ) replaceS3();
            Serializer serializer = getSerializer(bko.s3Key);
            putBytes(bko.bucketName,bko.s3Key,serializer.objectToByteArray(bko.obj));
        }

        private void putBytes(String bucketName,String key,byte[] b)
            throws S3ServiceException, IOException {

            S3Object o = null;
            try {
                o = new S3Object(key);
                o.setDataInputStream(new ByteArrayInputStream(b));
                o.setContentLength(b.length);
                o.setContentType(Mimetypes.MIMETYPE_OCTET_STREAM); // added 3/2/2010
                s3.putObject(bucketName,o);
            }
            finally {
                if ( o != null ) { o.closeDataInputStream(); o = null; }
            }
        }

        Serializer getSerializer(String s3Key) {
            if ( s3Key.endsWith("." + S3Ops.GZSER) ) {
                if ( gzipSerializer == null ) gzipSerializer = new GzipSerializer();
                return(gzipSerializer);
            }
//            else if ( s3Key.endsWith("." + S3Ops.JAR) ) return(standardSerializer);
            else {
                if ( standardSerializer == null ) standardSerializer = new StandardSerializer();
                return(standardSerializer);
            }
        }

        void replaceS3() throws S3ServiceException {
            if ( (s3 != null) && !s3.isShutdown() ) s3.shutdown();
            s3 = new RestS3Service(acs, null,null,jp);
        }
    }
}
